package xin.marcher.module.inventory.mq.consumer.listener;

import com.google.common.cache.Cache;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import xin.marcher.module.common.constants.inventory.TairInventoryConstant;
import xin.marcher.module.common.tair.TairLock;
import xin.marcher.module.common.utils.JsonUtil;
import xin.marcher.module.common.utils.SnowflakeIdWorker;
import xin.marcher.module.inventory.cache.InventoryBucketCache;
import xin.marcher.module.inventory.domain.bo.BucketLocalCache;
import xin.marcher.module.inventory.domain.request.InventoryRequest;

import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;

/**
 * 兜底处理 缓存的本地元数据同步刷新
 */
@Slf4j
@Component
public class BucketRefreshListener implements MessageListenerConcurrently {

    /**
     * 本地缓存
     */
    @Autowired
    private Cache cache;

    @Resource
    private InventoryBucketCache inventoryBucketCache;

    @Resource
    private TairLock tairLock;


    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
        try {
            for (MessageExt messageExt : list) {
                String msg = new String(messageExt.getBody());

                InventoryRequest inventoryRequest = JsonUtil.json2Object(msg, InventoryRequest.class);

                String key = inventoryRequest.getSellerId() + inventoryRequest.getSkuId();
                // 锁住这个商品的本地缓存同步，每次只会处理一个本地缓存元数据
                String value = SnowflakeIdWorker.getCode();
                boolean lock = tairLock.tryLock(TairInventoryConstant.SELLER_SYNC_BUCKET_PREFIX + key, value);
                if (lock) {
                    try {
                        String bucketLocalKey = TairInventoryConstant.SELLER_BUCKET_PREFIX + key;
                        // 远程缓存
                        BucketLocalCache bucketCache = inventoryBucketCache.getTairBucketCache(bucketLocalKey);
                        // 本地缓存
                        BucketLocalCache bucketLocalCache = inventoryBucketCache.getBucketLocalCache(key);

                        // 以远程缓存为准进行本地缓存覆盖(缓存最少需要一个分桶列表，远程缓存没有就需要保证增加一个可用分桶)
                        BucketLocalCache bucketLocalCache1 = inventoryBucketCache.diffRefreshCache(bucketLocalCache, bucketCache);
                        // 比较远程缓存和本地分桶是否一致
                        if (!Objects.isNull(bucketLocalCache1)) {
                            //覆盖本地缓存
                            cache.put(bucketLocalKey, bucketLocalCache1);
                            // 更新远程缓存
                            inventoryBucketCache.setBucketCache(bucketLocalKey, bucketLocalCache1);
                        }
                    } catch (Exception e) {
                        log.error("consume error, 同步刷新本地分桶元数据失败", e);
                        // 失败不重试
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } finally {
                        tairLock.unlock(key, value);
                    }
                }
            }
        } catch (Exception e) {
            log.error("consume error, 同步刷新本地分桶元数据失败", e);
            // 失败不重试
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
